package com.xian.java.stream;

import com.xian.bean.Temperature;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FilterTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //DataStreamSource<String> stringDataStreamSource = env.readTextFile("E:/test/Temperature.txt");
        DataStream<String> dataStream = env.readTextFile("E:/test/Temperature.txt"); //用父类接
        //dataStream.print();


        SingleOutputStreamOperator<Temperature> singleOutputStreamOperator = dataStream.map(new RichMapFunction<String, Temperature>() {
            @Override
            public Temperature map(String line) throws Exception {
                String[] arr = line.split(",");
                int id = Integer.parseInt(arr[0]);
                double t = Double.parseDouble(arr[1]);
                long ts = Long.parseLong(arr[2]);
                Temperature temperature = new Temperature(id, t, ts);
                return temperature;
            }
        });

//        singleOutputStreamOperator.print();

        SingleOutputStreamOperator<Temperature> singleOutputStreamOperator1 = singleOutputStreamOperator.filter(new RichFilterFunction<Temperature>() {
            @Override
            public boolean filter(Temperature t) throws Exception {
                return t.getId() > 90;
            }
        });

        singleOutputStreamOperator1.print();


        env.execute();


    }
}
